package com.jqj.example;

import java.time.LocalTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;

/**
 * @Desc:
 * @Author: lijingzheng
 * @Date: 2024/12/25
 */
public class ThreadPoolTest {

    public static void main1(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        CompletionService<String> completionService = new ExecutorCompletionService<>(executorService);

        List<Future<String>> futures = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            Future<String> submit = completionService.submit(() -> {
                int nextInt = new Random().nextInt(10);
                TimeUnit.SECONDS.sleep(nextInt);
                return String.format("%s ===>> %s ===>> %s", LocalTime.now(), nextInt, Thread.currentThread().getName());
            });
            futures.add(submit);
        }

        futures.parallelStream().forEach(future -> {
            try {
                String s = future.get();
                System.out.printf("\033[32;1m[%s ===>> %s] ===>> %s\033[0m\n", LocalTime.now(), Thread.currentThread(), s);
//                System.out.printf("\033[32;1m[%s ===>> %s] ===>> %s\033[0m\n", LocalTime.now(), Thread.currentThread(), future.get());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }

    public static void main2(String[] args) throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        CompletionService<String> completionService = new ExecutorCompletionService<>(executorService);

        for (int i = 0; i < 10; i++) {
            completionService.submit(() -> {
                int nextInt = new Random().nextInt(10);
                TimeUnit.SECONDS.sleep(nextInt);
                return String.format("%s ===>> %s ===>> %s", LocalTime.now(), nextInt, Thread.currentThread().getName());
            });
        }
        System.out.printf("\033[33;1m%s ===>> 任务提交完毕\033[0m\n\n", LocalTime.now());

        Future<String> future;
//        while ((future = completionService.take()) != null) { // 阻塞等待
//        while ((future = completionService.poll()) != null) { // 非阻塞
        while ((future = completionService.poll(2, TimeUnit.SECONDS)) != null) { // 阻塞等待超时
            String s = future.get();
            System.out.printf("\033[32;1m[%s ===>> %s] ===>> %s\033[0m\n", LocalTime.now(), Thread.currentThread(), s);
        }
    }

    public static void main(String[] args) throws Exception {
        ForkJoinPool forkJoinPool = new ForkJoinPool(4);
        MyRecursiveAction myRecursiveAction = new MyRecursiveAction(1024, "0");
        forkJoinPool.invoke(myRecursiveAction);

        System.in.read();
    }

    // RecursiveTask
    public static class MyRecursiveAction extends RecursiveAction {

        private final int workLoad;

        private String path;

        public MyRecursiveAction(int workLoad, String path) {
            this.workLoad = workLoad;
            this.path = path;
        }

        @Override
        protected void compute() {
            long poolQueuedTaskCount = ForkJoinTask.getPool().getQueuedTaskCount();
            long stealCount = ForkJoinTask.getPool().getStealCount();
            int threadQueuedTaskCount = ForkJoinTask.getQueuedTaskCount();
            //如果工作超出阈值，将任务分解成更小的任务
            if (this.workLoad > 10) {
                try {
                    TimeUnit.MILLISECONDS.sleep(128);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
                //将工作负载分成多个子任务
                List<MyRecursiveAction> subtasks = new ArrayList<>(createSubtasks(this.path));
                //将子任务加入到任务队列中
                for (RecursiveAction subtask : subtasks) {
                    subtask.fork();
                }
                System.out.printf("\033[34;1m%s ===>> %s ===>> %s ===>> %s ===>> 任务拆分 ===>> tqtc=%s ===>> " +
                                "sc=%s ===>> pqtc=%s\033[0m\n", LocalTime.now(), Thread.currentThread().getName(),
                        this.workLoad, this.path, threadQueuedTaskCount, stealCount, poolQueuedTaskCount);
            } else {
                int nextInt = new Random().nextInt(4) * this.workLoad;
                if (Thread.currentThread().getName().equals("ForkJoinPool-1-worker-1")) {
                    nextInt *= 3;
                }
                try {
                    TimeUnit.SECONDS.sleep(nextInt);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
                System.out.printf("\033[35;1m%s ===>> %s ===>> nextInt=%s ===>> workLoad=%s ===>> path=%s ===>> " +
                                "tqtc=%s ===>> sc=%s ===>> pqtc=%s\033[0m\n",
                        LocalTime.now(), Thread.currentThread().getName(), nextInt, this.workLoad, this.path,
                        threadQueuedTaskCount, stealCount, poolQueuedTaskCount);
            }
        }

        //将工作负载分成多个子任务
        private List<MyRecursiveAction> createSubtasks(String pPath) {
            List<MyRecursiveAction> subtasks = new ArrayList<>();
            MyRecursiveAction subtask1 = new MyRecursiveAction(this.workLoad / 2, pPath + "A");
            MyRecursiveAction subtask2 = new MyRecursiveAction(this.workLoad / 2, pPath + "B");
            subtasks.add(subtask1);
            subtasks.add(subtask2);
            return subtasks;
        }

    }

}
